import { snippets } from "@/lib/generated/snippets";
import { Snippet } from "@/components/code";
import { Callout, Card, Cards, Steps, Tabs } from "nextra/components";
import UniversalTabs from "@/components/UniversalTabs";

# Declarative Workflow Design (DAGs)

Hatchet workflows are designed in a **Directed Acyclic Graph (DAG)** format, where each task is a node in the graph, and the dependencies between tasks are the edges. This structure ensures that workflows are organized, predictable, and free from circular dependencies. By defining the sequence and dependencies of tasks upfront, you can easily understand the actual runtime state as compared to the expected state when debugging or troubleshooting.

## Defining a Workflow

Start by declaring a workflow with a name. The workflow object can declare additional workflow-level configuration options which we'll cover later.

The returned object is an instance of the `Workflow` class, which is the primary interface for interacting with the workflow (i.e. [running](./run-with-results.mdx), [enqueuing](./run-no-wait.mdx), [scheduling](./scheduled-runs.mdx), etc).

<UniversalTabs items={["Python", "Typescript", "Go"]}>
  <Tabs.Tab title="Python">

<Snippet src={snippets.python.dag.worker.define_a_dag} />

</Tabs.Tab>
<Tabs.Tab title="Typescript">

<Snippet src={snippets.typescript.dag.workflow.declaring_a_dag_workflow} />

</Tabs.Tab>
<Tabs.Tab title="Go">

<Snippet src={snippets.go.dag.main.declaring_a_workflow} />

</Tabs.Tab>
</UniversalTabs>

<Callout variant="info">
  The Workflow return object can be interacted with in the same way as a
  [task](./your-first-task.mdx), however, it can only take a subset of options
  which are applied at the task level.
</Callout>

## Defining a Task

Now that we have a workflow, we can define a task to be executed as part of the workflow. Tasks are defined by calling the `task` method on the workflow object.

The `task` method takes a name and a function that defines the task's behavior. The function will receive the workflow's input and return the task's output. Tasks also accept a number of other configuration options, which are covered elsewhere in our documentation.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
<Tabs.Tab title="Python">
  In Python, the `task` method is a decorator, which is used like this to wrap a function:

<Snippet src={snippets.python.dag.worker.first_task} />

The function takes two arguments: `input`, which is a Pydantic model, and `ctx`, which is the Hatchet `Context` object. We'll discuss both of these more later.

<Callout type="info">
In the internals of Hatchet, the task is called using _positional arguments_, meaning that you can name `input` and `ctx` whatever you like.

For instance, `def task_1(foo: EmptyModel, bar: Context) -> None:` is perfectly valid.

</Callout>

</Tabs.Tab>
<Tabs.Tab title="Typescript">

<Snippet src={snippets.typescript.dag.workflow.first_task} />

The `fn` argument is a function that takes the workflow's input and a
context object. The context object contains information about the workflow
run (e.g. the run ID, the workflow's input, etc). It can be synchronous or
asynchronous.

</Tabs.Tab>
<Tabs.Tab title="Go">

<Snippet src={snippets.go.dag.main.defining_a_task} />

</Tabs.Tab>
</UniversalTabs>

## Building a DAG with Task Dependencies

The power of Hatchet's workflow design comes from connecting tasks into a DAG structure. Tasks can specify dependencies (parents) which must complete successfully before the task can start.

<UniversalTabs items={["Python", "Typescript", "Go"]}>
<Tabs.Tab title="Python">

<Snippet src={snippets.python.dag.worker.task_with_parents} />

</Tabs.Tab>
<Tabs.Tab title="Typescript">

<Snippet src={snippets.typescript.dag.workflow.second_task_with_parent} />

</Tabs.Tab>
<Tabs.Tab title="Go">

<Snippet src={snippets.go.dag.main.adding_a_task_with_a_parent} />

</Tabs.Tab>
</UniversalTabs>

## Accessing Parent Task Outputs

As shown in the examples above, tasks can access outputs from their parent tasks using the context object:

<UniversalTabs items={["Python", "Typescript", "Go"]}>
<Tabs.Tab title="Python">

<Snippet src={snippets.python.dag.worker.task_with_parents} />

</Tabs.Tab>
<Tabs.Tab title="Typescript">

<Snippet src={snippets.typescript.dag.workflow.accessing_parent_outputs} />

</Tabs.Tab>
<Tabs.Tab title="Go">
```go
// Inside a task with parent dependencies
var parentOutput ParentOutputType
err := ctx.ParentOutput(parentTask, &parentOutput)
if err != nil {
    return nil, err
}
```
</Tabs.Tab>
</UniversalTabs>

## Running a Workflow

You can run workflows directly or enqueue them for asynchronous execution. All the same methods for running a task are available for workflows!

<UniversalTabs items={["Python", "Typescript", "Go"]}>
<Tabs.Tab title="Python">

<Snippet src={snippets.python.dag.trigger.trigger_the_dag} />

</Tabs.Tab>
<Tabs.Tab title="Typescript">

<Snippet
  src={snippets.typescript.simple.typed_run_methods.run_method_flavors}
/>

</Tabs.Tab>
<Tabs.Tab title="Go">

```go
// Run workflow and wait for the result
result, err := simple.Run(ctx, input)

// Enqueue workflow to be executed asynchronously
runID, err := simple.RunNoWait(ctx, input)
```

</Tabs.Tab>
</UniversalTabs>
